package com.ycy;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Properties;
import java.util.regex.Pattern;

public class Consumes {


  static Logger logger = LoggerFactory.getLogger(Produces.class);

  public static void main(String[] args) {
    Properties properties = new Properties();

    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Conf.IP + ":" + Conf.PORT); //ip
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

//    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "mmp");
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");


    KafkaConsumer kafkaConsumer = new KafkaConsumer(properties);
//    kafkaConsumer.subscribe(Collections.singletonList(Conf.TOPIC));

//    String topic1 = Conf.TOPIC1;
//    String topic2 = Conf.TOPIC2;
//
//    ArrayList<String> topics = new ArrayList<String>();
//    topics.add(topic1);
//    topics.add(topic2);
//    kafkaConsumer.subscribe(topics);

//    订阅pattern
    kafkaConsumer.subscribe(Pattern.compile("applog.*|fnw_.*|ycy.*"));
    while (true) {
      ConsumerRecords<String, String> record = kafkaConsumer.poll(Duration.ofSeconds(1));  //拉取消息，阻塞时间5秒
      if (record.isEmpty())
        continue;

//      遍历消息并打印value
      for (ConsumerRecord<String, String> rec : record) {
        logger.info("topic :{} key:{} val:{} offset:{}" ,rec.topic(),rec.key(),rec.value(),rec.offset());
      }

      kafkaConsumer.commitAsync();

    }

  }

}
